package com.zc.component.network.http.server.vertx;

import lombok.*;
import lombok.extern.slf4j.Slf4j;
import com.zc.core.topic.Topic;
import com.zc.component.network.core.DefaultNetworkType;
import com.zc.component.network.core.NetworkType;
import com.zc.component.network.http.server.HttpExchange;
import com.zc.component.network.http.server.HttpServer;
import org.springframework.http.HttpStatus;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Setter
@AllArgsConstructor
@NoArgsConstructor
@Slf4j
public class VertxHttpServer implements HttpServer {

    private Collection<io.vertx.core.http.HttpServer> httpServers;

    private HttpServerConfig config;

    private String id;

    private EmitterProcessor<HttpExchange> httpServerProcessor = EmitterProcessor.create(false);

    private FluxSink<HttpExchange> exchangeFluxSink = httpServerProcessor.sink();

    private final Topic<FluxSink<HttpExchange>> route = Topic.createRoot();

    @Getter
    @Setter
    private String lastError;

    @Setter(AccessLevel.PACKAGE)
    private InetSocketAddress bindAddress;

    public VertxHttpServer(HttpServerConfig config) {
        this.config = config;
        this.id = config.getId();
    }

    @Override
    public InetSocketAddress getBindAddress() {
        return bindAddress;
    }

    /**
     * 这段代码主要是为了将HTTP请求分发到已注册的处理逻辑或订阅者，并确保每个请求都能得到正确的响应处理。
     *
     * @param httpServers http 请求服务
     */
    public void setHttpServers(Collection<io.vertx.core.http.HttpServer> httpServers) {
        // 首先，如果当前服务已启动（isAlive() 返回 true），则会先调用 shutdown() 方法关闭现有的HTTP服务器。
        if (isAlive()) {
            shutdown();
        }
        // 然后将传入的 httpServers 集合赋值给类的成员变量 this.httpServers，这样可以管理并处理多个HTTP服务器。
        this.httpServers = httpServers;
        // 对于集合中的每一个 io.vertx.core.http.HttpServer 实例，为其设置一个请求处理器。当有新的HTTP请求到达时，这个处理器会被回调执行。
        for (io.vertx.core.http.HttpServer server : this.httpServers) {
            // requestHandler 方法是用来设置一个请求处理器（Handler），这个处理器将会处理所有接收到的HTTP请求
            // 当你调用 server.requestHandler(handler) 并传入一个 RequestHandler 实例时，该实例所包含的方法将被回调，
            // 每当有新的HTTP请求到达服务器时都会执行这个方法。这个方法通常会接受一个 HttpServerRequest 参数，这是Vert.x
            // 对HTTP请求的封装，包含了请求的所有相关信息，如请求方法、URL路径、请求头和请求体等。
            server.requestHandler(request -> {
                // 在处理器内部，首先处理可能发生的异常情况，将其记录到日志中。
                request.exceptionHandler(err -> {
                    log.error(err.getMessage(), err);
                });

                // 创建一个 VertxHttpExchange 对象，封装了当前的HTTP请求信息，以便进一步处理。
                VertxHttpExchange exchange = new VertxHttpExchange(request, config);

                // 通过解析请求URL和方法来查找对应的路由 (route.findTopic(...))，并获取该路由下的所有订阅者 (subscribers)
                String url = exchange.getUrl();
                if (url.endsWith("/")) {
                    url = url.substring(0, url.length() - 1);
                }

                // 如果找到订阅者，将 VertxHttpExchange 对象发送给每个订阅者进行处理，
                route.findTopic("/" + exchange.request().getMethod().name().toLowerCase() + url)
                    .flatMapIterable(Topic::getSubscribers)
                    .doOnNext(sink -> sink.next(exchange))
                    // 若未找到订阅者，则检查是否有默认的处理器，如果没有，则向客户端返回404 NOT FOUND状态码。
                    .switchIfEmpty(Mono.fromRunnable(() -> {
                        if (!httpServerProcessor.hasDownstreams()) {
                            log.warn("http server no handler for:[{} {}://{}{}]", request.method(), request.scheme(), request.host(), request.path());
                            request.response()
                                .setStatusCode(HttpStatus.NOT_FOUND.value())
                                .end();
                        }
                    }))
                    .subscribe();
                // 并且如果存在下游处理器（httpServerProcessor.hasDownstreams()返回true），
                // 也将 VertxHttpExchange 发送给全局的事件流(exchangeFluxSink.next(exchange)）。
                if (httpServerProcessor.hasDownstreams()) {
                    exchangeFluxSink.next(exchange);
                }
            });
        }
    }

    @Override
    public Flux<HttpExchange> handleRequest() {
        return httpServerProcessor;
    }


    /**
     *  整个方法的主要作用是根据给定的方法和URL模式配置路由，
     *  并为每个匹配的请求创建一个响应式的事件流，
     *  从而使得框架能够在请求到达时调用正确的处理器逻辑。
     * @param method     请求方法: {@link org.springframework.http.HttpMethod}
     * @param urlPatterns url
     * @return HTTP交换接口，支持获取请求和发送响应
     */
    @Override
    public Flux<HttpExchange> handleRequest(String method, String... urlPatterns) {
        // 使用了Reactor库中的 Flux.create 方法来创建一个反应式流，允许在订阅时动态生成事件。
        return Flux.create(sink -> {
            // 使用 Disposable.Composite 来管理所有订阅关系，确保在 Flux 被取消订阅时能够同时取消所有已添加的路由订阅。
            Disposable.Composite disposable = Disposables.composite();
            // 遍历传入的URL模式数组，对每个模式执行以下操作：
            for (String urlPattern : urlPatterns) {
                String pattern = Stream.of(urlPattern.split("[/]"))
                    .map(str -> {
                        //处理路径变量,如: /devices/{id}
                        // 将路径中的变量替换为通配符 *，以便于后续的路由匹配。
                        if (str.startsWith("{") && str.endsWith("}")) {
                            return "*";
                        }
                        return str;
                    })
                    .collect(Collectors.joining("/"));
                // 根据需要添加或删除前导/尾随斜杠以确保模式正确格式化。
                if (pattern.endsWith("/")) {
                    // 去掉 pattern 字符串末尾的 /。这个逻辑通常是用来规范化URL路径，确保不以斜杠结尾，因为在某些路由系统或文件路径表示中，
                    // 连续的尾部斜杠或者不必要的末尾斜杠可能会引起歧义或问题。通过移除末尾的 /，可以统一处理方式，并保持路径的一致性。
                    pattern = pattern.substring(0, pattern.length() - 1);
                }
                if (!pattern.startsWith("/")) {
                    pattern = "/".concat(pattern);
                }
                // 依据HTTP方法和处理后的URL模式构建一个新的路由路径，形如 /GET/devices/{id}。
                pattern = "/" + method + pattern;
                log.debug("handle http request : {}", pattern);
                // 调用 route.append(pattern) 添加新的路由监听器，并订阅返回的 Topic<FluxSink<HttpExchange>>。这样当有匹配此模式的请求到来时，对应的处理器会收到通知。
                Topic<FluxSink<HttpExchange>> sub = route.append(pattern);
                sub.subscribe(sink);
                // 使用 Disposable.Composite 来管理所有订阅关系，确保在 Flux 被取消订阅时能够同时取消所有已添加的路由订阅。
                disposable.add(() -> sub.unsubscribe(sink));
            }
            // 最后，设置 sink.onDispose(disposable)，使得当外部订阅者取消订阅这个 Flux 时，它会自动解除所有已注册的路由回调函数。
            sink.onDispose(disposable);
        });
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public NetworkType getType() {
        return DefaultNetworkType.HTTP_SERVER;
    }

    @Override
    public void shutdown() {
        if (httpServers != null) {
            for (io.vertx.core.http.HttpServer httpServer : httpServers) {
                httpServer.close(res -> {
                    if (res.failed()) {
                        log.error(res.cause().getMessage(), res.cause());
                    } else {
                        log.debug("http server [{}] closed", httpServer.actualPort());
                    }
                });
            }
            httpServers.clear();
            httpServers = null;
        }
    }

    @Override
    public boolean isAlive() {
        return httpServers != null && !httpServers.isEmpty();
    }

    @Override
    public boolean isAutoReload() {
        return false;
    }
}
